package ar.edu.itba.pod.legajo46470.event;

import static com.google.common.base.Preconditions.checkNotNull;

import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.HashSet;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;
import org.joda.time.DateTime;

import ar.edu.itba.event.EventInformation;
import ar.edu.itba.event.RemoteEventDispatcher;
import ar.edu.itba.node.Node;
import ar.edu.itba.node.NodeInformation;
import ar.edu.itba.pod.agent.runner.Agent;
import ar.edu.itba.pod.legajo46470.cluster.node.ClusterNode;
import ar.edu.itba.pod.legajo46470.communication.ClusterCommunication;
import ar.edu.itba.pod.thread.CleanableThread;

public class RemoteEventDispatcherImpl extends UnicastRemoteObject implements RemoteEventDispatcher {
	private static final long serialVersionUID = -3807009621486107663L;
	private static final long REMOVETIME = 5000;
	private static final long UPDATETIME = 5000;
	private static final double STOPGOSSIP = 0.05;
	private final static Logger LOGGER = Logger.getLogger(RemoteEventDispatcherImpl.class);
	private ConcurrentMap<EventInformation, Long> processedEvents;
	private BlockingQueue<EventInformation> pendingEvents;
	private ConcurrentMap<NodeInformation, Long> lastUpdates;
	private final ClusterNode clusterNode;
	private final ClusterCommunication<RemoteEventDispatcher> clusterComm;
	
	public RemoteEventDispatcherImpl(ClusterNode clusterNode) throws RemoteException {
		super();
		checkNotNull(clusterNode, "Cluster Node cannot be null");
		
		this.clusterNode = clusterNode;	
		this.pendingEvents = new LinkedBlockingQueue<EventInformation>();
		this.processedEvents = new ConcurrentHashMap<EventInformation, Long>();
		this.lastUpdates = new ConcurrentHashMap<NodeInformation, Long>();
		this.clusterComm = new ClusterCommunication<RemoteEventDispatcher>(clusterNode, Node.DISTRIBUTED_EVENT_DISPATCHER);
		initializeEventReader();
		initializeEventRemover();
		initializeEventUpdates(); // TODO NO ARRANCAR EL THREAD ACA PORQ NO ARRANCO LA SIMULACION!
	}
	
	@Override
	public boolean publish(EventInformation event) throws RemoteException,
			InterruptedException {
	// sync to prevent receiving events until moveQueueFor is executed!
		boolean newEvent = false;
		if (!(pendingEvents.contains(event) || processedEvents.containsKey(event))) {
			pendingEvents.add(event);			
			newEvent = true;
		}
		return newEvent;
	}

	@Override
	public Set<EventInformation> newEventsFor(NodeInformation nodeInformation)
			throws RemoteException {
		Set<EventInformation> newEvents = new HashSet<EventInformation>();
		Long lastUpdate, processedTime;

			
		for (EventInformation evinfo : processedEvents.keySet()) {
			lastUpdate = lastUpdates.get(nodeInformation);
			processedTime = processedEvents.get(evinfo);
			
			if (lastUpdate != null) {
				if (lastUpdate < processedTime) {
					newEvents.add(evinfo);
					lastUpdates.put(nodeInformation, DateTime.now().getMillis());
				}
			}
		}	
		
		return newEvents;
	}

	@Override
	public BlockingQueue<Object> moveQueueFor(Agent agent)
			throws RemoteException {
			return ((DistributedEventDispatcher)(this.clusterNode.getLocalEventDispatcher())).deregister(agent);
	}
	
	private void initializeEventReader() {
		Thread readTask = new CleanableThread("EventReader") {
			public void run() {
				Random gossiping = new Random();
				try {
					while (true) {
						EventInformation eventInfo;
						eventInfo = pendingEvents.take();														
						
						
						Set<NodeInformation> neighbours;
						RemoteEventDispatcher remoteStubDispatcher;
						boolean newGossip;
						
						
						if (clusterNode.getLocalEventDispatcher() instanceof DistributedEventDispatcher) {
							DistributedEventDispatcher innerDispatcher = (DistributedEventDispatcher)clusterNode.getLocalEventDispatcher();
							
							// Event comes from outside
							if (!eventInfo.nodeId().equals(clusterNode.getNodeInfo().id())) {
								innerDispatcher.publishInternal(eventInfo.source(), eventInfo.event());
							}

							// event has been processed
							Long updateTime = DateTime.now().getMillis();
							
							// TODO ver que objeto usar como sincro
							synchronized (processedEvents) {
								processedEvents.put(eventInfo, updateTime);
							}
							
							neighbours = clusterComm.getRandomNeighbours();
							// gossip broadcast
							for (NodeInformation n : neighbours) {
								try {
									remoteStubDispatcher = clusterComm.getStub(n);
								
									newGossip = remoteStubDispatcher.publish(eventInfo);
									lastUpdates.put(n, updateTime);
									
									if (!newGossip) {
										// Gossip known, with certain probability, stop
										// transmiting gossip!
										if (gossiping.nextFloat() < STOPGOSSIP) {
											break;
										}
									}
								} catch (RemoteException e) {
									// Unreacheable node
									LOGGER.warn("Node down: " + n.id());
									clusterComm.disconnectNode(n);
								} catch (IllegalStateException e) {
									// Unreacheable node, disconnected inside getStub
									LOGGER.warn("Node down: " + n.id());
								}
							}
						
						} else {
							throw new IllegalArgumentException();
						}
					}
				} catch (InterruptedException e) {
					LOGGER.warn("Event Reader Thread finished!");
				} catch (IllegalArgumentException e) {
					LOGGER.error("Incorrect type of local dispatcher");
				}
			};
		};
		
		readTask.start();
	}
	
	private void initializeEventRemover() {
		Thread removeTask = new CleanableThread("EventCleaner") {
			
			public void run() {
				try {
					while(true) {
						Thread.sleep(REMOVETIME);
					
						for (EventInformation evinfo : processedEvents.keySet()) {
							if (isOld(processedEvents.get(evinfo), DateTime.now().getMillis())) {
								processedEvents.remove(evinfo);
							}
						}
					}
				} catch (InterruptedException e) {
					LOGGER.info("EventRemover: sleep interrupted");
				}
			};
			
			private boolean isOld(Long processedTime, Long currentTime) {
				if (currentTime - processedTime > REMOVETIME) {
					return true;
				} else {
					return false;
				}
			};
		};
		
		removeTask.start();
	}
	
	private void initializeEventUpdates() {
		Thread updatesTask = new CleanableThread("EventUpdates") {
			Set<NodeInformation> someNeighbours;
			Set<EventInformation> updates;
			RemoteEventDispatcher remoteEventDispatcher;
			public void run() {
				try {
					// TODO AGREGAR EN TODOS LOS THREADS Q TIENEN WHILE TRUE un !finish que sea una variable
					// del objeto DistributedSimulation que se setee en false cuando se hace simulation.stop()
					while(true) {
						Thread.sleep(UPDATETIME);
						
						// always return a non null Set, buy may be empty as well
						someNeighbours = clusterComm.getRandomNeighbours();
						for (NodeInformation n : someNeighbours) {
							try {
								remoteEventDispatcher = (RemoteEventDispatcher) clusterComm.getStub(n);
								updates = remoteEventDispatcher.newEventsFor(clusterNode.getNodeInfo());
								
								if (updates != null && updates.size() > 0) {
									pendingEvents.addAll(updates);
								}
							} catch (IllegalStateException e) {
								LOGGER.info("Node down: " + n.id());
								clusterComm.disconnectNode(n);
							} catch (RemoteException e) {
								LOGGER.info("Node down: " + n.id());
								clusterComm.disconnectNode(n);
							}
						}
					}
				} catch (InterruptedException e) {
					LOGGER.info("EventRemover: sleep interrupted");
				}
			};
		};
		
		updatesTask.start();
	}
}

